#include "zqlog_plugin.h"
#include <signal.h>
#include <thread>
#include <iostream>
#include "zqlog.h"

namespace libzq
{

	void ZQConsolePlugin::init(const CSimpleIniA& ini)
	{
		set_color(ZQLOG_White);
	}

	void ZQConsolePlugin::emit(ZQLogLevel log_level, const string& log_msg)
	{
		set_color(level_color(log_level));
		std::cout << log_msg << std::endl;
		set_color(ZQLOG_White);
	}

	void ZQConsolePlugin::destory()
	{
		set_color(ZQLOG_White);
	}

	libzq::ZQLogColor ZQConsolePlugin::level_color(ZQLogLevel log)
	{
		switch (log)
		{
		case libzq::ZQLog_Trace:
			return ZQLOG_White;
			break;
		case libzq::ZQLog_Debug:
			return ZQLOG_Yellow;
			break;
		case libzq::ZQLog_Info:
			return ZQLOG_Green;
			break;
		case libzq::ZQLog_Warn:
			return ZQLOG_Pink;
			break;
		case libzq::ZQLog_Error:
		case libzq::ZQLog_Fatal:
			return ZQLOG_Red;
			break;
		default:
			return ZQLOG_White;
			break;
		}
	}

	void ZQConsolePlugin::set_color(ZQLogColor color)
	{
#ifdef WIN32
		HANDLE hConsole = GetStdHandle(STD_OUTPUT_HANDLE);
		SetConsoleTextAttribute(hConsole, color);
#else
		printf("\033[%d;40;1m", color);
#endif
	}

	void ZQFilePlugin::init(const CSimpleIniA& ini)
	{
		_stop = false;
		_log_stream.open("./log.txt", std::ios_base::app);
		_thread = std::thread(std::bind(&ZQFilePlugin::run, this));
	}

	void ZQFilePlugin::emit(ZQLogLevel log_level, const string& log_msg)
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_log_msg = log_msg;
		_cond_var.notify_one();
	}

	void ZQFilePlugin::destory()
	{
		_stop = true;
		std::unique_lock<std::mutex> lock(_mutex);
		_cond_var.notify_one();
		_thread.join();
	}

	void ZQFilePlugin::run()
	{
		std::unique_lock<std::mutex> lock(_mutex);
		while (!_stop)
		{
			_cond_var.wait(lock);
			_log_stream << _log_msg << "\n";
			_log_stream.flush();
		}
	}

	string ZQFilePlugin::current_date()
	{
		return "";
	}

#ifdef ZQLOG_PLUGIN_KAFKA

	void ZQKafkaPlugin::init(const CSimpleIniA& ini)
	{
		int partition = ini.GetLongValue("kafka", "partition", RD_KAFKA_PARTITION_UA);
		const char* brokers = ini.GetValue("kafka", "brokers", "localhost:9092");
		const char* topic = ini.GetValue("kafka", "topic", "zqlog");
		init_kafka(partition, brokers, topic);
	}

	void ZQKafkaPlugin::emit(ZQLogLevel log_level, const string& log_msg)
	{
		int ret = rd_kafka_produce(_topic, _partition, RD_KAFKA_MSG_F_COPY, (void*)log_msg.c_str(), (size_t)log_msg.length(), NULL, 0, NULL);
		rd_kafka_poll(_handler, 0);
	}

	void ZQKafkaPlugin::destory()
	{
		/* Destroy topic */
		rd_kafka_topic_destroy(_topic);

		/* Destroy the handle */
		rd_kafka_destroy(_handler);
	}

	void ZQKafkaPlugin::init_kafka(int partition, const char* brokers, const char* topic)
	{
		char tmp[16] = { 0 };
		char errstr[512] = { 0 };

		/* Kafka configuration */
		_conf = rd_kafka_conf_new();

		//set logger :register log function  
		//rd_kafka_conf_set_log_cb(conf_, logger);

		/* Quick termination */
		snprintf(tmp, sizeof(tmp), "%i", SIGIO);
		rd_kafka_conf_set(_conf, "internal.termination.signal", tmp, NULL, 0);

		/*topic configuration*/
		_topic_conf = rd_kafka_topic_conf_new();

		if (!(_handler = rd_kafka_new(RD_KAFKA_PRODUCER, _conf, errstr, sizeof(errstr))))
		{
			LOGE("Failed to create new producer: %s", errstr);
			return;
		}
		//rd_kafka_set_log_level(handler_, LOG_DEBUG);

		/* Add brokers */
		if (rd_kafka_brokers_add(_handler, brokers) == 0)
		{
			LOGE("No valid brokers specified");
			return;
		}

		/* Create topic */
		_topic = rd_kafka_topic_new(_handler, topic, _topic_conf);
		LOGI("Success create kafka topic [%s]", topic);
		return;
	}

#endif
}